package com.coolingme.consumer.nonauto;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * 消费者客户端，手动非同步提交offset
 * <p>
 * 虽然同步提交 offset 更可靠一些，但是由于其会阻塞当前线程，直到提交成功。
 * 因此吞吐量会收到很大的影响。因此更多的情况下，会选用异步提交 offset 的方式。
 *
 * @author wangyue
 * @date 2021/7/16 9:48
 */
public class ConsumerAsyncDemo {

    public static void main(String[] args) {

        // 配置信息
        Properties properties = new Properties();
        // 连接的集群
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");

        // key,value反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        // 消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "coolingme");
        // 更换一个新的消费者组
//        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "coolingme2");

        // 重置消费者的offset
        // 当前消费者组不存在或服务上没有数据（如：数据被删除）
        // earliest:重置为最早的offset，获取最早的数据
        // latest:重置为最新的offset，获取最新的数据
//        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 消费者
        KafkaConsumer<String, Object> consumer = new KafkaConsumer<>(properties);
        // 订阅主题,若主题不存在会自动创建
        consumer.subscribe(Arrays.asList("test-topic", "coolingme"));
        while (true) {
            // 拉取数据
            ConsumerRecords<String, Object> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, Object> record : records) {
                System.out.println(String.format("key= %s ,value= %s ,topic= %s ,partition= %d ,offset= %d", record.key(), record.value(), record.topic(), record.partition(), record.offset()));
            }
            // 异步提交
            consumer.commitAsync((offsets, exception) -> {
                if (exception != null) {
                    System.err.println("Commit	failed	for " + offsets);
                }
            });
        }

    }
}

